﻿using System;

using Confluent.Kafka;

using RpcClient.Interface;
using RpcClient.Kafka.Interface;

using RpcHelper;

using Wedonek.Gateway.Modular.Interface;

namespace Wedonek.Gateway.Modular.Service
{
        [RpcClient.Attr.ClassLifetimeAttr(RpcClient.Attr.ClassLifetimeType.单例)]
        internal class KafkaDemo : IKafkaDemo
        {
                private readonly IKafkaService _Kafka = null;
                private IKafkaConsumer _Consumer;
                private string[] _Topic = new string[] { "Demo_Queue" };
                public KafkaDemo(IKafkaService kafka)
                {
                        this._Kafka = kafka;
                        if (_Kafka.IsInit)
                        {
                                _Consumer = this._Kafka.Subscrib<string> (this._Subscrib);
                        }
                }

                public void Dispose()
                {
                        _Consumer?.Dispose();
                }

                public void Producer()
                {
                        if (!this._Kafka.Producer("Demo_Msg", _Topic))
                        {
                                throw new ErrorException("demo.kafka.send.fail");
                        }
                }
                public void Subscribe()
                {
                        _Consumer.AddBind(_Topic);
                        _Consumer.Subscribe();
                }

                private void _Subscrib(IKafkaEvent<Ignore, string> obj)
                {
                        Console.WriteLine("Kafka_Topic：" + obj.Topic);
                        Console.WriteLine("收到Kafka订阅消息：" + obj.Value);
                }
        }
}
